package cn.dmp.report

import cn.dmp.utils.AdRptKpi
import org.apache.commons.lang.StringUtils
import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext

/**
  * 媒体分析报表
  *     spark broadcast
  */
object RptAppAnalysis {

    Logger.getLogger("org.apache").setLevel(Level.WARN)

    def main(args: Array[String]): Unit = {


        // 检验参数 -> 配置文件
        if (args.length != 3) {
            println(
                """
                  |cn.dmp.report.RptAppAnalysis
                  |参数：dataInputPath, dictInputPath, outputPath
                """.stripMargin)
            sys.exit()
        }

        val Array(dataInputPath, dictInputPath, outputPath) = args

        // sparkcontext
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("媒体分析")
        val sc = new SparkContext(sparkConf)

        // 读取数据 - parquet
        val sqlc = new SQLContext(sc)
        // 数据文件 - dataframe
        val parquet = sqlc.read.parquet(dataInputPath)


        // app_dict.txt -> 通过爬虫从网站上爬去的
        val appdict = sc.textFile(dataInputPath).map(line => line.split("\t", -1))
          .filter(_.length >= 5)
          .map(t => (t(4), t(1)))
          .collect().toMap
        // 广播字典数据
        val broadcast = sc.broadcast(appdict)

        // driver 错误代码示例
        /*val appDictMap = collection.mutable.HashMap[String, String]()
        sc.textFile(dataInputPath).map(line => line.split("\t", -1))
          .filter(_.length >= 5)
          .foreach(arr => {
              appDictMap.put(arr(4), arr(1)) // executor
          })
        val broadcast = sc.broadcast(appDictMap)
        println("=============="+appDictMap.size)*/


        // 统计
        parquet.map(row => {

            val appId = row.getAs[String]("appid")
            var appName = row.getAs[String]("appname")

            // 如果appName为空，则根据appId -》 app_dict中找对应的appName,如果没找到就用appId作为appName
            // appName和appId都为空，将appName=未知
            if (StringUtils.isEmpty(appName)) {
                if (StringUtils.isNotEmpty(appId)) {
                    appName = broadcast.value.getOrElse(appId, appId)
                } else appName = "未知"
            }

            (appName, AdRptKpi(row)._2)
        }).reduceByKey((list1, list2) => list1.zip(list2).map(t => t._1+ t._2))
          .foreach(println)
          //.saveAsTextFile(outputPath)

        sc.stop()

    }

}
